package com.ciccwm.log.kafka;

import com.ciccwm.log.util.TraceIdUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.util.StringUtils;

import java.util.Map;

/**
 * Kafka生产者链路追踪拦截器.
 * 用于在发送Kafka消息时，将当前线程的追踪ID添加到消息头中.
 * 支持在分布式系统中追踪消息的完整发送链路.
 */
@Slf4j
public class TraceIdProducerInterceptor<K, V> implements ProducerInterceptor<K, V> {
    /**
     * 在发送消息时处理追踪ID.
     * 将当前线程的追踪ID添加到消息头中.
     *
     * @param record 生产者记录
     * @return 处理后的生产者记录
     */
    @Override
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
        String traceId = TraceIdUtil.getTraceId();
        if (StringUtils.hasText(traceId)) {
            record.headers().add(TraceIdUtil.TRACE_ID_KEY, traceId.getBytes());
            log.debug("Added trace ID to Kafka producer record: {}", traceId);
        }

        return record;
    }

    /**
     * 在消息确认时处理.
     * 当前实现不执行任何操作.
     *
     * @param metadata 记录元数据
     * @param exception 异常信息
     */
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        // Do nothing
    }

    /**
     * 关闭拦截器.
     * 当前实现不执行任何操作.
     */
    @Override
    public void close() {
        // Do nothing
    }

    /**
     * 配置拦截器.
     * 当前实现不执行任何操作.
     *
     * @param configs 配置参数
     */
    @Override
    public void configure(Map<String, ?> configs) {
        // Do nothing
    }
} 
